package com.etlmaster.executor.steps.loop;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;

import org.springframework.jdbc.support.rowset.SqlRowSet;
import org.springframework.jdbc.support.rowset.SqlRowSetMetaData;

import com.etlmaster.executor.utils.StepState;
import com.etlmaster.jdbc.JdbcTemplate;

/**
 * @author Larry
 * @copyright Larry@etlMaster
 * @since 2017-09-10
 */
public class SqlLoopProcessor extends LoopProcessor {

	@Override
	public int process() throws InterruptedException, ExecutionException {
		int posi = expression.indexOf(" in ");
		String variableName = expression.substring(0, posi);
		String[] sqlStatementArr = expression.substring(posi + 4,
				expression.length()).split("#");
		String sqlStatement = sqlStatementArr[0];
		Object rowSet = null;
		if (job.getVariables().containsKey(sqlStatement)) {
			rowSet = job.getVariables().get(sqlStatement);
		} else {
			String dataSource = "";
			if (sqlStatementArr.length > 1) {
				dataSource = sqlStatementArr[1];
			}
			JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource);
			rowSet = jdbcTemplate.queryForRowSet(sqlStatement);
		}

		if (rowSet instanceof SqlRowSet) {
			SqlRowSet sqlRowSet = (SqlRowSet) rowSet;
			SqlRowSetMetaData srsmd = sqlRowSet.getMetaData();
			int colSize = srsmd.getColumnCount();

			while (sqlRowSet.next()) {
				i++;
				Map<String, Object> rowData = new HashMap<String, Object>();
				for (int j = 0; j < colSize; j++) {
					String colName = srsmd.getColumnName(j + 1);
					rowData.put(colName.toLowerCase(),
							sqlRowSet.getObject(colName));
				}
				job.putVariable(variableName, rowData);
				Boolean isBreak = (Boolean) job.getVariable("isBreak");
				if ((isBreak != null && (Boolean) job.getVariable("isBreak"))
						|| loopStep instanceof EndLoopStep) {
					break;
				}

				loopStepMeta.runLoop(job,i,"i");

			}
		} else if (rowSet instanceof List) {
			List list = (List) rowSet;

			for (int j = 0; j < list.size(); j++) {
				Object rowData = list.get(j);
				Boolean isBreak = (Boolean) job.getVariable("isBreak");
				if ((isBreak != null && (Boolean) job.getVariable("isBreak"))
						|| loopStep instanceof EndLoopStep) {
					break;
				}
				
				loopStepMeta.runLoop(job,i,"i");


			}
		}
		loopStepMeta.startLoopStack();

		return StepState.RUNNING_SUCCESSED;

	}

}
